package com.wwz.config;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.wwz.feign.ProductFeign;
import com.wwz.feign.UserFeign;
import com.wwz.service.OrderService;
import com.wwz.tcc.TccOrderService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/*@Component
@RabbitListener(queues = "single_queue")*/
public class RabbitReceiverTwo {
    private static Logger LOGGER = LoggerFactory.getLogger(RabbitReceiverTwo.class);
    @Autowired
    private ProductFeign productFeign;
    @Autowired
    private UserFeign userFeign;
    @Autowired
    private OrderService orderService;
    @Autowired
    private TccOrderService tccOrderService;
    @Resource(name="masterStringRedisTemplate")
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisLuaTemplate redisLuaTemplate;
    private static AtomicInteger atomicInteger = new AtomicInteger(0);
    private static ExecutorService executorService = new ThreadPoolExecutor(10,20,5000L, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(10));
    private static AnnotationConfigApplicationContext applicationContext  = new AnnotationConfigApplicationContext();

    /**
     * 默认情况下,如果没有配置手动ACK, 那么Spring Data AMQP 会在消息消费完毕后自动帮我们去ACK
     * 存在问题：如果报错了,消息不会丢失,但是会无限循环消费,一直报错,如果开启了错误日志很容易就吧磁盘空间耗完
     * 解决方案：手动ACK,或者try-catch 然后在 catch 里面将错误的消息转移到其它的系列中去
     * spring.rabbitmq.listener.simple.acknowledge-mode = manual
     */
    @RabbitHandler
    /*@GlobalTransactional*/
    public void cfgUserReceiveDealy(Long orderId, Message message, Channel channel) throws Exception {
        LOGGER.info("===============接收队列接收消息RabbitReceiver====================");
        LOGGER.info("接收时间:{},接受内容:{},参数属性:{}", LocalDateTime.now(), orderId.toString(),message.getMessageProperties());
        //通知 MQ 消息已被接收,可以ACK(从队列中删除)了
        boolean isAck = true;
        try {
            channel.basicQos(0);
            long branchId = System.currentTimeMillis();
            String id = stringRedisTemplate.opsForValue().get("order:keyprefix:orderId");
            if(!org.apache.commons.lang3.StringUtils.isEmpty(id)&&Integer.parseInt(id)>=1000){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("name","5");
            jsonObject.put("branch_id",branchId);
            jsonObject.put("num",5);
            if(!redisLuaTemplate.increment("orders:keyprefix:orderId",jsonObject.toJSONString(),"order:keyprefix:"+branchId,-1,"10000000")){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            };
          /*  executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    applicationContext.publishEvent(new ApplicationEvent(new String("我发布的事件")){});
                }
            });*/

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            isAck = false;
            LOGGER.error("============消费失败,尝试消息补发再次消费!==============");
            LOGGER.error(e.getMessage());
            try{
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }catch (Exception ee){
                LOGGER.error("===========回退失败!==============");
            }
            throw new Exception(e.getMessage());
        }
    }
}
